
from datetime import datetime, timedelta

import pandas as pd
import numpy as np

from air_web.web_flask.tools.air_base import AirBase


class ZBData(AirBase):
    URLS_MAP = {"zb_curve_data": "zhixiang/addExchRtCurveDataOrg",
                "zb_index_data": "zhixiang/addAirconditionIndex"}

    def __init__(self, log, today_str):
        super().__init__(log, today_str=today_str)
        self.yesterday = (datetime.strptime(self.today_str, "%Y-%m-%d") - timedelta(
            days=1)).strftime("%Y-%m-%d")
        self.tomorrow = (datetime.strptime(self.today_str, "%Y-%m-%d") + timedelta(
            days=1)).strftime("%Y-%m-%d")
        self.tomorrow_end = datetime.strptime(self.tomorrow, "%Y-%m-%d") + timedelta(days=1)
        self.res_list = []

    def proc_and_save_index_data(self, res_df, data_date):
        res_df['data_date'] = data_date
        res_df['index_org_no'] = "51101"
        res_df['index_p_org_no'] = '00000'
        res_df['index_id'] = res_df['index_org_no'] + '_' + res_df['index_no'] + '_' + res_df[
            'data_date'].str.replace("-", "")
        res_df["index_cons_no"] = None
        res_df["index_code"] = None

        res_df = res_df.replace({np.nan: None})
        res_list = res_df.to_dict("records")
        self.res_list.extend(res_list)

    def proc_ctrl_data_time(self, df):
        df["time_str"] = df["data_time"].dt.strftime('%H:%M')
        df = df.loc[((df["time_str"] >= "12:00") & (df["time_str"] < "14:00")) |
                    ((df["time_str"] >= "20:00") & (df["time_str"] < "22:00"))]
        df.drop("time_str", axis=1, inplace=False)
        return df

    def get_ctrl_curve_data(self, curve_type, p_field, ctrl_org_list, data_date, is_tomorrow=False):
        end_date = self.tomorrow if is_tomorrow else self.today_end
        self.log.info(f"start proc curve_type: {curve_type}")
        sql = f"""select data_time, sum({p_field})/10000 as p_value
                  from orgno_typeid_15min
                  where data_time>='{self.today_str}'
                    and data_time < '{end_date}'
                    and org_no in ({','.join(ctrl_org_list)})
                    and type_id=1
                  group by data_time
                  order by data_time
              """
        df = self.sql_engine.query(sql)
        if df.empty:
            self.log.warning(f"orgno_typeid_15min未查询到{self.today_str}数据:p_field:{p_field}")
            return
        if is_tomorrow:
            df["data_time"] = df["data_time"] + timedelta(days=1)
        df = self.proc_ctrl_data_time(df)
        self.proc_and_save_curve_data(df, curve_type, data_date)

    def get_ctrl_org_list(self, data_date):
        sql = f"""select c.org_no, r.p_org_no, org_level
                  from city_approval_info c
                  left join real_org_no r on c.org_no=r.org_no
                  where control_time <= '{data_date}'
                    and end_time >= '{data_date}'
               """
        approval_df = self.sql_engine.query(sql)
        if approval_df.empty:
            self.log.warning(f"{data_date}无调控地区")
            return []

        # 去重
        if len(approval_df['org_level'].drop_duplicates()) == 2:
            on7_parent_list = approval_df.loc[approval_df['org_level'] == 2,
                                              "p_org_no"].drop_duplicates().tolist()
            on5_list = approval_df.loc[approval_df['org_level'] == 1, "org_no"].tolist()

            on7_parent_dupl = list(set(on5_list) & set(on7_parent_list))
            if on7_parent_dupl:
                approval_df = approval_df[~approval_df['p_org_no'].isin(on7_parent_dupl)]

        org_list = approval_df["org_no"].astype(str).tolist()
        self.log.info(f"{self.today_str}调控地区:{org_list}")
        return org_list

    @staticmethod
    def get_point_time_df(data_date):
        time_list = pd.date_range(data_date, periods=96, freq="15T")
        point_time_df = pd.DataFrame(
            {
                "data_time": time_list,
                "point": [f"point{str(i)}" for i in range(1, 97)],
            }
        )
        return point_time_df

    def proc_and_save_curve_data(self, df, curve_type, data_date):
        df = df.round(6)
        point_time_df = self.get_point_time_df(data_date)
        if len(df) < 96:
            df = df.merge(point_time_df, on="data_time", how="right")

        df['data_date'] = data_date
        pro_df = df.groupby("data_date")['p_value'].apply(list).reset_index()
        p_cols = point_time_df['point'].tolist()
        pro_df[p_cols] = pd.DataFrame(pro_df['p_value'].values.tolist())

        all_cols = ['data_date'] + p_cols
        res_df = pro_df[all_cols]
        res_df['curve_org_no'] = "51101"
        res_df['curve_p_org_no'] = "00000"
        res_df['curve_tgt_type'] = '21'
        res_df['curve_type'] = curve_type
        res_df['curve_id'] = res_df['curve_org_no'] + '_' + res_df['curve_tgt_type'] + '_' + res_df[
            'curve_type'] + '_' + res_df['data_date'].str.replace("-", "")

        # 字段
        res_df["curve_cons_no"] = None
        res_df["stats_freq"] = 1

        res_df = res_df.replace({np.nan: None})
        res_list = res_df.to_dict("records")
        self.res_list.extend(res_list)

    def get_curve_data(self, curve_type, p_field):
        self.log.info(f"start proc curve_type: {curve_type}")
        sql = f"""select data_time, {p_field}/10000 as p_value
                  from orgno_typeid_15min
                  where data_time>='{self.today_str}'
                    and data_time < '{self.today_end}'
                    and org_no='51101'
                    and type_id=1
                  order by data_time
              """
        df = self.sql_engine.query(sql)
        if df.empty:
            self.log.warning(f"orgno_typeid_15min未查询到{self.today_str}数据:p_field:{p_field}")
            return
        self.proc_and_save_curve_data(df, curve_type, self.today_str)

    def tmp_index_main(self):
        """(晚上推送第二天)南瑞要求改成凌晨推送当天的指标"""
        self.log.info("start #### tmp_index_data")

        sql = f"""select max(tmp) as max_tmp, min(tmp) as min_tmp
                  from aclr_weather_hour
                  where datetime >= '{self.today_str}'
                    and datetime < '{self.today_end}'
               """
        tmp_df = self.sql_engine.query(sql)
        tmp_df = tmp_df.dropna(axis=0, how='any')
        if tmp_df.empty:
            self.log.warning(f"aclr_weather_hour未查询到{self.today_str}数据")
            return
        else:
            tmp_df = tmp_df.round(0).astype("int64")
            tmp_dict = tmp_df.to_dict("records")[0]
            res_dict = {'index_no': ['10602010', '10602020'],
                        'index_name': ['最高温度', '最低温度'],
                        'index_value': [tmp_dict['max_tmp'], tmp_dict['min_tmp']]}
            res_df = pd.DataFrame(res_dict)
            self.proc_and_save_index_data(res_df, self.today_str)

            self.log.debug(f"tmp_index_data:{self.res_list}")
            self.post_data(self.URLS_MAP["zb_index_data"], self.res_list)

    def cons_index_main(self):
        """凌晨推送当天的指标:用户数"""
        self.log.info("start #### cons_index_data")
        sql = f"""select count(distinct cons_no) as count
                   from aclr_base_doc_all
                   where type_code_sort = 1
              """
        cons_num_df = self.sql_engine.query(sql)
        if cons_num_df.empty:
            self.log.warning(f"aclr_base_doc_all未查询到数据")
            return
        else:
            cons_num = cons_num_df['count'].sum()

        index_df = pd.DataFrame([{"index_no": "10601030", "index_name":"自主调节接入用户数量",
                                  "index_value": cons_num}])
        self.proc_and_save_index_data(index_df, self.today_str)

        self.log.debug(f"cons_index_data:{self.res_list}")
        self.post_data(self.URLS_MAP["zb_index_data"], self.res_list)

    def static_main(self):
        """晚上推送明天调控计划的数据:基线是今天"""
        self.log.info("start #### zb_static_curve_data")
        ctrl_org_list = self.get_ctrl_org_list(self.tomorrow)
        if ctrl_org_list:
            self.get_ctrl_curve_data("311", "p_kt_sum", ctrl_org_list, self.tomorrow,
                                     is_tomorrow=True)  # 事件调控-空调调控基线负荷
            self.get_ctrl_curve_data("312", "p_kt_sum * 0.75", ctrl_org_list, self.tomorrow,
                                     is_tomorrow=True)  # 事件调控-空调调控目标负荷

        self.log.debug(f"zb_static_curve_data:{self.res_list}")
        #self.post_data(self.URLS_MAP["zb_curve_data"], self.res_list)

    def rt_main(self):
        """实时数据推送"""
        self.log.info("start #### zb_rt_curve_data")
        self.get_curve_data("304", "p_total_sum")  # 304 聚合用户关口负荷
        self.get_curve_data("308", "p_kt_sum")   # 308 聚合自主调节用户空调负荷
        self.get_curve_data("321", "p_kt_sum * 0.25")  # 321 自主调节可调负荷

        ctrl_org_list = self.get_ctrl_org_list(self.today_str)
        #if ctrl_org_list:
            #self.get_ctrl_curve_data("309", "p_total_sum", ctrl_org_list, self.today_str)
            #self.get_ctrl_curve_data("310", "p_kt_sum", ctrl_org_list, self.today_str)

        self.log.debug(f"zb_rt_curve_data:{self.res_list}")
        self.post_data(self.URLS_MAP["zb_curve_data"], self.res_list)
